package com.catmiao.spark.stream

import com.catmiao.spark.stream.SparkStreaming12_Req1_BlackList.AdClickData
import com.catmiao.spark.util.JDBCUtil
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * @title: SparkStreaming01_WordCount
 * @projectName spark_study
 * @description: 最近十分钟广告点击量
 * @author ChengMiao
 * @date 2024/3/25 00:31
 */
object SparkStreaming12_Req3 {

  def main(args: Array[String]): Unit = {


    // 创建环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    //  param1 : 环境配置，SparkConf
    //  param2 ： 采集周期【批量处理周期】
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //3.定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->
        "localhost:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "test",
      "key.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" ->
        "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //4.读取 Kafka 数据创建 DStream
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String](ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Set("first"), kafkaPara))


    /**
     * 1. 开窗确定时间范围；
     * 2. 在窗口内将数据转换数据结构为((adid,hm),count);
     * 3. 按照广告 id 进行分组处理，组内按照时分排序。
     */
    val sdf: SimpleDateFormat = new SimpleDateFormat("HH:mm")
    val dataDs: DStream[AdClickData] = kafkaDStream.map(
      data => {
        val d: String = data.value()
        val value = d.split(" ")
        AdClickData(value(0), value(1), value(2), value(3), value(4))
      }
    )

    // 设定窗口 为10min 每次步长为1min
    val windowsDs: DStream[AdClickData] = dataDs.window(Minutes(10), Minutes(1))

    val value: DStream[((String, String), Int)] = windowsDs.map(
      item => {
        val date = sdf.format(new Date(item.ts.toLong))
        ((item.ad, date), 1)
      }
    )

    // 统计窗口内 各个广告每分钟的总数
    val windowCountDs: DStream[((String, String), Int)] = value.reduceByKey(_ + _)

    // 转换结构
    val v2: DStream[(String, (String, Int))] = windowCountDs.map {
      case ((ad, date), count) => {
        (ad, (date, count))
      }
    }

    // 按广告id分组
    val result: DStream[(String, List[(String, Int)])] = v2.groupByKey().mapValues(
      iter => {
        // 按时间排序
        iter.toList.sortWith(_._1 < _._1)
      }
    )


    result.print()



    ssc.start()
    ssc.awaitTermination()
  }


}
